#include "thread_pool.h"

/**
  * @brief  线程取消处理函数
  * @note   防止线程在上了锁之后，突然被取消了，所以在此处执行解锁操作
  * @param  arg:  需要解锁那个线程
  * @retval None
  */ 
void THREAD_POOL_Handler(void *arg)
{
    printf("[%u] is ended.\n", (unsigned)pthread_self());
	pthread_mutex_unlock((pthread_mutex_t *)arg);   
}

/**
  * @brief  线程任务函数
  * @note   None
  * @param  arg:  需要传进线程池管理结构体指针，方便管理
  * @retval None
  */ 
void* THREAD_POOL_Routine(void *arg)
{
    // 打印相关调试信息
    #ifdef DEBUG
	    printf("[%u] is started.\n", (unsigned)pthread_self()); // 打印线程的TID
    #endif

    // 一、获取线程池结构体的地址,因为条件变量和互斥锁都在该结构体定义
	thread_pool *pool = (thread_pool *)arg;

	// 二、定义任务指针,指向要提取的任务
	struct task *p;

    // 三、
    while (1)
	{
		/*
            ** push a cleanup functon handler(), make sure that
            ** the calling thread will release the mutex properly
            ** even if it is cancelled during holding the mutex.
            **
            ** NOTE:
            ** pthread_cleanup_push() is a macro which includes a
            ** loop in it, so if the specified field of codes that
            ** paired within pthread_cleanup_push() and pthread_
            ** cleanup_pop() use 'break' may NOT break out of the
            ** truely loop but break out of these two macros.
            ** see line 61 below.
		*/
        //================================================//
		// 注册取消处理函数,&pool->lock 把锁传递过去，当前线程突然被取消时，防止死锁
		pthread_cleanup_push(THREAD_POOL_Handler, (void *)&pool->lock);

		// 上锁 ,因为所有线程都是在同一个任务列表中提取任务,用互斥锁保护链表中的任务
		pthread_mutex_lock(&pool->lock);
		//================================================//

        // no task, and is NOT shutting down, then wait
        // 1、没有任务，并且线程池没有关机，那么就等待
		while (pool->waiting_tasks == 0 && !pool->shutdown)
		{
			pthread_cond_wait(&pool->cond, &pool->lock); // 如果没有任何任务，则进入等待状态
		}
        
		// no task, and is shutting down, then exit
        // 2、没有任务，但是线程池是关机的，那么直接退出此线程
		if (pool->waiting_tasks == 0 && pool->shutdown == true)
		{
			pthread_mutex_unlock(&pool->lock);
			pthread_exit(NULL); // CANNOT use 'break'; //如果没有任务且是关机状态则，退出线程
		}

        // have some task, then consume it 
        // 3、如果有任务，那么运行它
		p = pool->task_list->next;		    // 指向第一个任务节点
		pool->task_list->next = p->next;    // 把任务节点从链表提取出来
		pool->waiting_tasks--;			    // 任务数量减少

        //================================================//
		pthread_mutex_unlock(&pool->lock);  // 解锁 ，因为已经取完了任务
		pthread_cleanup_pop(0);
		//================================================//

        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); // 关闭取消请求,确保任务被执行完毕
		(p->do_task)(p->arg);								  // 执行任务并传递参数 void *(*do_task)(void *arg);
        // p->do_task(p->arg);
		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);  // 任务执行完毕后重启开启取消请求

        free(p); // 释放任务节点
    }

    pthread_exit(NULL);
}

/**
  * @brief  初始化线程池结构体
  * @note   None
  * @param  pool:           线程池管理结构体指针	
  *         threads_number：一开始有多少个线程(员工) 
  * @retval 成功：true
  *         失败：false
  */ 
 bool THREAD_POOL_Init(thread_pool* pool, unsigned int threads_number)
 {
    // 1、初始化互斥锁、条件变量
    pthread_mutex_init(&pool->lock, NULL);
	pthread_cond_init(&pool->cond, NULL);

    // 2、初始化线程池管理结构体
    pool->shutdown = false;                                         // 线程池一开始处于开机阶段

	pool->task_list = malloc(sizeof(struct task));                  // 给任务链表结构体申请堆空间
	pool->tids = malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS);    // 给线程id申请空间,要多少个就申请多少个，由MAX_ACTIVE_THREADS决定
    if (pool->task_list == NULL || pool->tids == NULL)              // 如果任务链表和线程id申请空间失败，则直接退出此函数        
	{
		perror("allocate memory error\n");
		return false;
	}

	pool->task_list->next = NULL;                                   // 给任务链表结构体赋处置，它下一个结构体指向NULL
	
    pool->max_waiting_tasks = MAX_WAITING_TASKS;                    // 最大设定的任务数            
	pool->waiting_tasks     = 0;                                    // 正在等待的任务数
	pool->active_threads    = threads_number;                       // 一开始公司的员工数

    // 3、申请线程函数
    int i;
	for (i = 0; i < pool->active_threads; i++)                      // 不断地运行出线程来
	{
		if (pthread_create(&((pool->tids)[i]), NULL,
						   THREAD_POOL_Routine, (void *)pool) != 0)
		{
			perror("create threads error\n");
			return false;
		}

        #ifdef DEBUG                                                // 调试信息，需要打印的调试信息的时候，编译时加这个标志即可，如：gcc main.c -o main -DDEBUG
                printf("[%u]:[%s] ==> tids[%d]: [%u] is created.\n",
                    (unsigned)pthread_self(), __FUNCTION__,
                    i, (unsigned)pool->tids[i]);
        #endif
	}

	return true;

 }

 /**
  * @brief  给线程池投放任务函数
  * @note   None
  * @param  pool:    线程池管理结构体指针	
  *         do_task：投放的线程任务函数(客户)
  *         arg：    传递给任务函数的参数
  * @retval 成功：true
  *         失败：false
  */ 
bool THREAD_POOL_AddTask(thread_pool *pool, void* (do_task)(void* arg), void *arg)
{
    // 1、新建一个任务节点
	struct task *new_task = malloc(sizeof(struct task));
	if (new_task == NULL)
	{
		perror("allocate memory error\n");
		return false;
	}

    // 2、初始化任务节点
	new_task->do_task = do_task;        // 将任务链表中的任务指针指向任务函数
	new_task->arg = arg;                // 传给任务函数的参数
	new_task->next = NULL;              // 因为是单向链表的尾插法，所以新节点指向的下一个为NULL

    // 上锁 ,防止在添加任务的时候，有别的线程去拿任务
	//============ LOCK =============//
	pthread_mutex_lock(&pool->lock);
	//===============================//

    // 3、如果当前等待的任务 大于 最大任务数
	if (pool->waiting_tasks >= MAX_WAITING_TASKS)
	{
		pthread_mutex_unlock(&pool->lock); // 解锁
		fprintf(stderr, "too many tasks.\n");
		free(new_task); // 放弃任务
		return false;	// 结束函数
	}

    // 4、将指针移动任务链表的末尾
    struct task *tmp = pool->task_list;
	while (tmp->next != NULL)
		tmp = tmp->next; // 偏移链表末尾

    // 5、尾插法
    tmp->next = new_task;  // 尾插
	pool->waiting_tasks++; // 任务数量增加


    //=========== UNLOCK ============//
	pthread_mutex_unlock(&pool->lock); // 任务已经添加完毕了，可以解锁了
	//===============================//

#ifdef DEBUG
	printf("[%u][%s] ==> a new task has been added.\n",
		   (unsigned)pthread_self(), __FUNCTION__);
#endif

    // 6、通知正在休眠的线程干活！
    pthread_cond_signal(&pool->cond); 

}

 /**
  * @brief  增加线程(招聘员工)
  * @note   None
  * @param  pool:               线程池管理结构体指针	
  *         additional_threads：要增加的线程数
  * @retval 成功：返回成功创建的线程数
  *         失败：小于等于0的整数(其中返回0代表没有添加任何线程)
  */ 
int THREAD_POOL_AddThread(thread_pool *pool, unsigned int additional_threads)
{
    // 1、如果添加的线程数为0，则退出
    if(additional_threads == 0)
        return 0;
    
    // 2、计算出现在需要的线程数
    // 总线程数 7                 当前的 5             +  添加的 2
	unsigned total_threads = pool->active_threads + additional_threads;
	int i, actual_increment = 0;

	// 3、循环创建新的任务线程 5              7
	for (i = pool->active_threads; i < total_threads && i < MAX_ACTIVE_THREADS; i++)
	{
		if (pthread_create(&((pool->tids)[i]), NULL, THREAD_POOL_Routine, (void *)pool) != 0)
		{
			perror("add threads error");

			// no threads has been created, return fail
			if (actual_increment == 0)
				return -1;

			break;
		}
		// 增加创建成功的线程数
		actual_increment++;

        #ifdef DEBUG
            printf("[%u]:[%s] ==> tids[%d]: [%u] is created.\n",
                (unsigned)pthread_self(), __FUNCTION__,
                i, (unsigned)pool->tids[i]);
        #endif
    }

    // 4、将成功添加的线程，进行记录
    pool->active_threads += actual_increment;

    // 5、返回成功创建的线程数
	return actual_increment;
}

  /**
  * @brief  减少线程(解雇员工)
  * @note   None
  * @param  pool:             线程池管理结构体指针	
  *         removing_threads：要删除的线程数
  * @retval 成功：返回还剩下的线程数
  *         失败：false(其中返回0代表没有删除任何线程)
  */ 
int THREAD_POOL_RemoveThread(thread_pool *pool, unsigned int removing_threads)
{
    // 1、如果删除的线程数为0，则退出
    if(removing_threads == 0)
        return 0;

    // 2、计算删除后还应该剩余的线程数
	//              剩余3      = 当前的5  -  需要删除的2
	int remaining_threads = pool->active_threads - removing_threads;

	// 如果剩余的 小于 0 ,则保留一个线程,执行任务
	remaining_threads = remaining_threads > 0 ? remaining_threads : 1;

	int i;              
	for (i = pool->active_threads - 1; i > remaining_threads - 1; i--)
	{
		// 循环取消线程
		errno = pthread_cancel(pool->tids[i]);
		if (errno != 0)
			break;
            
        #ifdef DEBUG
            printf("[%u]:[%s] ==> cancelling tids[%d]: [%u]...\n",
                (unsigned)pthread_self(), __FUNCTION__,
                i, (unsigned)pool->tids[i]);
        #endif
	}    

    // 3、返回剩下的活跃线程
	if (i == pool->active_threads - 1)
		return -1;
	else
	{
		pool->active_threads = i + 1;
		return i + 1;
	}

}


  /**
  * @brief  删除线程池资源(解散公司)
  * @note   None
  * @param  pool:    线程池管理结构体指针	
  * @retval 成功：true
  *         失败：false
  */ 
bool THREAD_POOL_Destroy(thread_pool *pool)
{

    // 1|设置关机标记为真
	pool->shutdown = true; 

    // activate all threads
    // 2、广播条件变量，唤醒所有休眠的线程
	pthread_cond_broadcast(&pool->cond);

    // 2, wait for their exiting
	int i;
	for (i = 0; i < pool->active_threads; i++)
	{
		errno = pthread_join(pool->tids[i], NULL); // 循环回收线程资源
		if (errno != 0)
		{
			printf("join tids[%d] error: %s\n",
				   i, strerror(errno));
		}
		else
			printf("[%u] is joined\n", (unsigned)pool->tids[i]);
	}

    // 3, free memories 释放所有的堆空间
	free(pool->task_list);
	free(pool->tids);
	free(pool);

	return true;

}